package cn.jbooter.mq.autoconfigure.rocketmq.coms;

import java.util.UUID;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.Assert;

import cn.jbooter.coms.exceptions.FrameworkException;
import cn.jbooter.mq.autoconfigure.rocketmq.properties.RocketProperties;

/**
 * rocket默认的生产者
 * @author HeJian
 */
public class RocketDefaultProducer implements IRocketProducer {
	private static final Logger logger = LoggerFactory.getLogger(RocketDefaultProducer.class);
	@Autowired
	private RocketProperties rocketProperties;
	private DefaultMQProducer producer = null;
	/**默认使用类名**/
	private String beanName = this.getClass().getSimpleName();

	public String getBeanName() {
		return beanName;
	}
	public void setBeanName(String beanName) {
		this.beanName = beanName;
	}

	@Override
	public void init() throws Exception {
		//参数校验
		Assert.notNull(this.rocketProperties,"rocketmq配置信息为空!");
		Assert.hasText(this.rocketProperties.getNamesrvAddr(), "rocketmq name server address未配置");
		Assert.hasText(this.rocketProperties.getMqProducerGroup(), "rocketmq未配置mq-producer-group");
		
		//初始化producer
		DefaultMQProducer producer = new DefaultMQProducer(this.rocketProperties.getMqProducerGroup());
        producer.setNamesrvAddr(this.rocketProperties.getNamesrvAddr());
        producer.setSendMsgTimeout(this.rocketProperties.getSendMsgTimeout());
        producer.setSendMessageWithVIPChannel(this.rocketProperties.isVipChannelEnabled());
        producer.setInstanceName(UUID.randomUUID().toString());
		producer.start();
		this.producer=producer;
		logger.info("rocketmq producer : {} is started.",this.beanName);
	}

	@Override
	public void destroy() throws Exception {
		if(this.producer!=null) {
			logger.info("shutdown rocketmq producer : {}",this.beanName);
			this.producer.shutdown();
		}
	}

	@Override
	public DefaultMQProducer getProducer() {
		if(producer==null) {
			 try {
				init();
			} catch (Exception e) {
				logger.error("初始化producer失败:"+e.getMessage());
				e.printStackTrace();
				throw new FrameworkException(e.getMessage());
			}
		}
		return this.producer;
	}

}
